package com.bw.dws;

import com.bw.bean.UserBehavior;
import com.bw.util.ProfileCalcUtil;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.math.BigDecimal;
import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Map;

/**
 * 性别标签实时计算（品类行为得分）
 * 文档参考：🔶1-109（性别定义）、🔶1-129（行为权重）、🔶1-152（家庭用户判定）
 */
public class GenderTagCalc extends KeyedProcessFunction<String, UserBehavior, Map<String, Object>> {
    // 1. 性别相关品类映射（文档🔶1-119/1-123/1-124）
    private static final Map<String, String> GENDER_CATEGORY = new HashMap<>();
    static {
        // 女性品类
        GENDER_CATEGORY.put("女装/女士精品", "female");
        GENDER_CATEGORY.put("女鞋", "female");
        GENDER_CATEGORY.put("美容护肤/美体/精油", "female");
        // 男性品类
        GENDER_CATEGORY.put("男装", "male");
        GENDER_CATEGORY.put("流行男鞋", "male");
        GENDER_CATEGORY.put("网游装备/游戏币/帐号/代练", "male");
        // 家庭品类
        GENDER_CATEGORY.put("家居用品", "family");
        GENDER_CATEGORY.put("厨房电器", "family");
        GENDER_CATEGORY.put("儿童用品", "family");
    }

    // 2. 状态存储：女性/男性/家庭得分 + 购买总权重
    private ValueState<BigDecimal> femaleScoreState;
    private ValueState<BigDecimal> maleScoreState;
    private ValueState<BigDecimal> familyScoreState;
    private ValueState<BigDecimal> totalBuyWeightState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 状态TTL=30天（文档"近30天行为"）
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.days(30)).build();

        femaleScoreState = getRuntimeContext().getState(new ValueStateDescriptor<>("femaleScore", BigDecimal.class));
        maleScoreState = getRuntimeContext().getState(new ValueStateDescriptor<>("maleScore", BigDecimal.class));
        familyScoreState = getRuntimeContext().getState(new ValueStateDescriptor<>("familyScore", BigDecimal.class));
        totalBuyWeightState = getRuntimeContext().getState(new ValueStateDescriptor<>("totalBuyWeight", BigDecimal.class));

        // 初始化状态
        if (femaleScoreState.value() == null) femaleScoreState.update(BigDecimal.ZERO);
        if (maleScoreState.value() == null) maleScoreState.update(BigDecimal.ZERO);
        if (familyScoreState.value() == null) familyScoreState.update(BigDecimal.ZERO);
        if (totalBuyWeightState.value() == null) totalBuyWeightState.update(BigDecimal.ZERO);
    }

    @Override
    public void processElement(UserBehavior behavior, Context ctx, Collector<Map<String, Object>> out) throws Exception {
        // （1）获取当前状态
        BigDecimal femaleScore = femaleScoreState.value();
        BigDecimal maleScore = maleScoreState.value();
        BigDecimal familyScore = familyScoreState.value();
        BigDecimal totalBuyWeight = totalBuyWeightState.value();

        // （2）计算行为权重
        double behaviorWeight = ProfileCalcUtil.getBehaviorWeight(behavior.getBehaviorType());
        BigDecimal weight = BigDecimal.valueOf(behaviorWeight);

        // （3）累加对应品类得分
        String genderType = GENDER_CATEGORY.getOrDefault(behavior.getCategoryName(), "other");
        switch (genderType) {
            case "female":
                femaleScore = femaleScore.add(weight);
                break;
            case "male":
                maleScore = maleScore.add(weight);
                break;
            case "family":
                familyScore = familyScore.add(weight);
                // 累加购买总权重（仅buy行为）
                if ("buy".equals(behavior.getBehaviorType())) {
                    totalBuyWeight = totalBuyWeight.add(weight);
                }
                break;
        }

        // （4）更新状态
        femaleScoreState.update(femaleScore);
        maleScoreState.update(maleScore);
        familyScoreState.update(familyScore);
        totalBuyWeightState.update(totalBuyWeight);

        // （5）判定性别标签（文档🔶1-152/1-155逻辑）
        String genderTag;
        // 家庭用户判定：家庭购买占比≥30% 且 男女得分差≤10%
        BigDecimal familyBuyRatio = totalBuyWeight.compareTo(BigDecimal.ZERO) == 0 ? 
                BigDecimal.ZERO : familyScore.divide(totalBuyWeight, 4, BigDecimal.ROUND_HALF_UP);
        BigDecimal scoreDiff = femaleScore.subtract(maleScore).abs();
        BigDecimal tenPercent = femaleScore.add(maleScore).multiply(new BigDecimal("0.1"));

        if (familyBuyRatio.compareTo(new BigDecimal("0.3")) >= 0 && scoreDiff.compareTo(tenPercent) <= 0) {
            genderTag = "家庭用户";
        } else if (femaleScore.compareTo(maleScore.add(tenPercent)) > 0) {
            genderTag = "女性";
        } else if (maleScore.compareTo(femaleScore.add(tenPercent)) > 0) {
            genderTag = "男性";
        } else {
            genderTag = "未知";
        }

        // （6）输出结果
        Map<String, Object> result = new HashMap<>();
        result.put("userId", behavior.getUserId());
        result.put("genderTag", genderTag);
        result.put("updateTime", new Timestamp(System.currentTimeMillis()));
        out.collect(result);
    }
}